package edu.zju.gis.dbfg.fileserver.dataLoader;

import com.google.gson.Gson;
import edu.zju.gis.dbfg.common.exception.LoaderException;
import edu.zju.gis.dbfg.fileserver.service.ConstructedDatasetService;
import edu.zju.gis.dbfg.fileserver.util.BeanUtil;
import edu.zju.gis.dbfg.common.util.ValidateUtils;
import edu.zju.gis.dbfg.fileserver.config.ESConfig;
import edu.zju.gis.dbfg.fileserver.constant.ConstructedDataType;

import edu.zju.gis.dbfg.fileserver.constant.ConstructedDataFieldType;
import edu.zju.gis.dbfg.fileserver.constant.FieldFormatType;
import edu.zju.gis.dbfg.fileserver.model.ConstructedDataField;
import edu.zju.gis.dbfg.fileserver.util.DateUtil;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.*;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.geotools.geojson.geom.GeometryJSON;
import org.json.JSONException;
import org.json.JSONObject;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.geom.GeometryFactory;
import org.locationtech.jts.io.ParseException;
import org.locationtech.jts.io.WKTReader;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;


/**
 * @author Hu
 * @date 2019/6/10
 * 接入 ConstructedDataset, 实现指定源的Dataset以及对应元数据的ES入库
 **/
@Getter
@Setter
public class ESLoader extends Loader {
    private ConstructedDatasetService constructedDatasetService;

    private ESConfig conf;

    private TransportClient client;

    private String indexName;

    /**
     * ES 单条数据入库
     *
     * @param value
     */
    @Override
    public void load(Map<ConstructedDataField, String> value) {
        String id = getId(value);
        String json = mapToJson(value);
        // TODO type是表名，还有点纠结，一个新的个人空间数据集是要用一个新的index来管理，还是用一个新的mapping来管理
        // TODO 毕竟 es 7.x 把 type  去掉了
        String type = "type";
        IndexResponse response = client.prepareIndex(this.indexName, type, id)
                .setSource(json, XContentType.JSON).get();
        if (!response.getResult().equals(DocWriteResponse.Result.CREATED)) {
            StringBuffer sb = new StringBuffer();
            Gson gson = new Gson();
            sb.append(gson.toJson(value) + "\n");
            sb.append(response.getResult().toString());
            errors.add(sb.toString());
        }
    }

    /**
     * ES 数据批量入库
     *
     * @param values
     */
    @Override
    public int batchLoad(List<Map<ConstructedDataField, String>> values) {
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        String type = this.dataset.getName();
        int batch_success = 0;
        for (Map<ConstructedDataField, String> value : values) {
            String id = getId(value);
            String json = mapToJson(value);
            bulkRequest.add(client.prepareIndex(this.indexName, type, id)
                    .setSource(json, XContentType.JSON));
        }
        BulkResponse bulkResponse = bulkRequest.get();
        if (bulkResponse.hasFailures()) {
            BulkItemResponse[] responses = bulkResponse.getItems();
            batch_success = values.size();
            for (int i = 0; i < responses.length; i++) {
                BulkItemResponse response = responses[i];
                if (response.isFailed()) {
                    batch_success--;
                    StringBuffer sb = new StringBuffer();
                    sb.append("\nLoad Error:[")
                            .append(i)
                            .append("]: index [")
                            .append(response.getIndex())
                            .append("], type [")
                            .append(response.getType())
                            .append("], id [")
                            .append(response.getId())
                            .append("], message [")
                            .append(response.getFailureMessage())
                            .append("]");
                    errors.add("\nImport Error: Es:" + sb.toString());
                }
            }
        } else {
            batch_success = values.size();
        }
        return batch_success;
    }

    @Override
    public Boolean validate(Map<ConstructedDataField, String> eachRow) {
        for (ConstructedDataField field : eachRow.keySet()) {
            //TODO 数据验证
            if (field.getIsTime()) {
                if (!ValidateUtils.isDate(eachRow.get(field))) {
                    errors.add("\nValidate Error: DataTime Format is Wrong:" + field.getName() + ":" + eachRow.get(field));
                    return false;
                }
            }
            if (field.getType().equals(ConstructedDataFieldType.Integer.name()) || field.getType().equals(ConstructedDataFieldType.Long.name())) {
                if (!ValidateUtils.isInteger(eachRow.get(field))) {
                    errors.add("\nValidate Error: Not Integer:" + field.getName() + ":" + eachRow.get(field));
                    return false;
                }
            }
            if (field.getType().equals(ConstructedDataFieldType.Float.name()) || field.getType().equals(ConstructedDataFieldType.Double.name())) {
                if (!ValidateUtils.isDecimal(eachRow.get(field))) {
                    errors.add("\nValidate Error: Not Float:" + field.getName() + ":" + eachRow.get(field));
                    return false;
                }
            }
        }
        return true;
    }

    // TODO 对于特殊字段，如datetime以及geometry(lat,lon,start_lat,start_lon)等特殊处理
    @Override
    public List<Map<ConstructedDataField, String>> mapValues(List<String> lines) throws IOException {
        this.total += lines.size();
        List<Map<ConstructedDataField, String>> result = new ArrayList<>();
        List<ConstructedDataField> metaFields = this.getMetaFields();
        // 循环遍历数据，将数据转换为dataset定义的字段中
        // TODO 默认以逗号或tab分隔，需要提供多种类型支撑
        String regex = lines.get(0).contains("\t") ? "\t" : ",";
        CSVFormat format = CSVFormat.DEFAULT.withDelimiter(regex.charAt(0));

        for (String line : lines) {
            Map<ConstructedDataField, String> tuple = new HashMap<>();
            CSVParser csvParser = CSVParser.parse(line, format);
            CSVRecord eachline = csvParser.getRecords().get(0);
            String[] srcFields = new String[eachline.size()];
            for (int i = 0; i < srcFields.length; i++) {
                srcFields[i] = eachline.get(i);
            }
            for (ConstructedDataField field : metaFields) {
                tuple.put(field, this.getFormatValue(field, srcFields));
            }
            result.add(tuple);
        }

        return result;
    }

    // TODO 这些和index管理的功能之后放到IndexManager类中
    private boolean createIndex() {
        if (this.indexExist()) {
            errors.add("\nImport Warming: :" + "ES Index already exists" + indexName);
            return true;
        }
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        //TODO number_of_shards 是否需要修改
        CreateIndexResponse response = indicesAdminClient.prepareCreate(indexName)
                .setSettings(Settings.builder()
                        .put("index.number_of_shards", 3)
                        .put("index.number_of_replicas", 0)).get();
        if (!response.isAcknowledged()) {
            throw new LoaderException(String.format("Create index %s fail", indexName));
        }
        String test = datasetToEsMapping();
        System.out.println(test);
        AcknowledgedResponse putResponse = indicesAdminClient.preparePutMapping(indexName)
                .setType(this.dataset.getName())
                .setSource(datasetToEsMapping(), XContentType.JSON).get();
        if (!putResponse.isAcknowledged()) {
            throw new LoaderException("Put mapping to index" + indexName + " failed");
        }
        return true;
    }

    /**
     * 数据装载过程初始化
     * 包括初始化后端参数设置、初始化client、初始化Loader参数、新建索引、装载数据源等操作
     *
     * @return
     */
    @Override
    public void init() {
        super.init();
        errors = new ArrayList<>();
        this.conf = BeanUtil.getBean(ESConfig.class);
        this.constructedDatasetService = BeanUtil.getBean(ConstructedDatasetService.class);
        initClient();
        this.indexName = dataset.getId();
        createIndex();
        setSource();
    }

    /**
     * 装载过程结束，进行清理
     */
    @Override
    public void finish() {
        if (this.client != null) {
            this.client.close();
        }
        if (this.source != null) {
            try {
                source.close();
            } catch (Exception e) {
//                logger.error("Close data source fail");
                errors.add("\nImport Error: :Close data source fail");
            }
        }
        try {
            this.dataset.setSize(this.success);
            this.constructedDatasetService.update(this.dataset);
        } catch (Exception e) {
            errors.add("\nUpdate Error: :Fail to update the size/storePath.");
        }
    }

    private void initClient() {
        if (this.client == null) {
            Settings settings = Settings.builder()
                    .put("cluster.name", conf.getClusterName()).build();
            try {
                this.client = new PreBuiltTransportClient(settings)
                        .addTransportAddress(new TransportAddress(InetAddress.getByName(conf.getHostName()), conf.getTransportPort()));
//                        .addTransportAddress(new TransportAddress(InetAddress.getByName(conf.getHostName2()), conf.getTransportPort()))
//                        .addTransportAddress(new TransportAddress(InetAddress.getByName(conf.getHostName3()), conf.getTransportPort()));
            } catch (UnknownHostException e) {
                e.printStackTrace();
                throw new LoaderException("Unvalid host name with [" + conf.getHostName() + "]");
            }
        }
    }

    private boolean indexExist() {
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        IndicesExistsResponse response = indicesAdminClient.prepareExists(indexName).get();
        return response.isExists();
    }

    private String datasetToEsMapping() {
        List<ConstructedDataField> fields = this.getMetaFields();
        Map<String, Object> properties = new HashMap<>();
        for (ConstructedDataField field : fields) {
            Map<String, String> fieldMap = new HashMap<>();
            if (field.getName().equals(FieldFormatType.SPATIAL_FIELD_NAME)) {
                fieldMap.put("type", "geo_shape");
            } else if (field.getIsTime()) {
                fieldMap.put("type", "date");
                fieldMap.put("format", "epoch_millis");
            } else if (field.getIsChinese()) {
                fieldMap.put("analyzer", "ik_max_word");
                fieldMap.put("search_analyzer", "ik_smart");
            } else if (field.getType().equals(ConstructedDataFieldType.String.name())) {
                fieldMap.put("type", "text");
            } else if (field.getType().equals(ConstructedDataFieldType.Integer.name())) {
                fieldMap.put("type", "integer");
            }else if(field.getType().equals(ConstructedDataFieldType.Long.name())){
                fieldMap.put("type", "long");
            }
            else if (field.getType().equals(ConstructedDataFieldType.Float.name())) {
                fieldMap.put("type", "float");
            } else if (field.getType().equals(ConstructedDataFieldType.Double.name())) {
                fieldMap.put("type", "double");
            } else if (field.getType().equals(ConstructedDataFieldType.Boolean.name())) {
                fieldMap.put("type", "boolean");
            } else {
                throw new LoaderException("Unsupported Field Type");
            }
            properties.put(field.getName(), fieldMap);
        }
        Map<String, Object> mapping = new HashMap<>();
        mapping.put("properties", properties);
        Gson gson = new Gson();
        return gson.toJson(mapping);
    }

    private String mapToJson(Map<ConstructedDataField, String> value) {
        JSONObject json = new JSONObject();
        try {
            for (ConstructedDataField field : value.keySet()) {
                if (field.getType().equals(ConstructedDataFieldType.String.name())) {
                    json.put(field.getName(), value.get(field));
                    if (field.getName().equals(FieldFormatType.SPATIAL_FIELD_NAME)) {
                        String wkt = this.getWktGeometry(dataset.getDatatype(), field.getName(), value);
                        WKTReader reader = new WKTReader();
                        try {
                            Geometry geometry = reader.read(wkt);
                            GeometryJSON g = new GeometryJSON(9);
                            String geojson = g.toString(geometry);
                            JSONObject geo = new JSONObject(geojson);
                            json.put(field.getName(), geo);
                        } catch (ParseException | JSONException e) {
                            errors.add("\nParse Json Error: value: " + value.get(field) + " :" + field.getName());
                            return "";
                        }
                    }
                } else if (field.getType().equals(ConstructedDataFieldType.Integer.name())) {
                    json.put(field.getName(), Integer.valueOf(value.get(field)));
                } else if (field.getType().equals(ConstructedDataFieldType.Float.name())) {
                    json.put(field.getName(), Float.valueOf(value.get(field)));
                } else if (field.getType().equals(ConstructedDataFieldType.Double.name())) {
                    json.put(field.getName(), Double.valueOf(value.get(field)));
                } else if (field.getType().equals(ConstructedDataFieldType.Long.name())) {
                    json.put(field.getName(), Long.valueOf(value.get(field)));
                } else if (field.getType().equals(ConstructedDataFieldType.Boolean.name())) {
                    json.put(field.getName(), Boolean.valueOf(value.get(field)));
                } else {
                    errors.add("\nUnsupported FIELD_TYPE Error:" + value.get(field) + ":" + field);
                    return "";
                }
            }
        } catch (Exception e) {
            errors.add("\nUnKnown Error:" + e.getMessage());
            return "";
        }
        return json.toString();
    }

    private String getId(Map<ConstructedDataField, String> values) {
        for (ConstructedDataField field : values.keySet()) {
            if (field.getIsKey()) return values.get(field);
        }
        return UUID.randomUUID().toString();
    }

    private String getFormatValue(ConstructedDataField field, String[] srcFields) {
        if (field.getIsTime()) {
            try {
                String format = field.getFormat();
                String value = srcFields[header.get(field.getName())];
                return DateUtil.format(DateUtil.parse(value, format));
            } catch (java.text.ParseException e) {
                errors.add("\nFormat Parse Warning: Time Format is Wrong:" + field.getName() + srcFields[header.get(field.getName())]);
                return "";
            }
        }
        return srcFields[header.get(field.getName())];
    }

    private String getWktGeometry(String spatialType, String fieldName, Map<ConstructedDataField, String> eachRow) {
        switch (ConstructedDataType.valueOf(spatialType)) {
            case Point:
                return getPointWKT(eachRow);
            case Polygon:
                return getPolygonWKT(eachRow);
            case LineString:
                return getPolyLineWKT(eachRow);
        }
        errors.add("\nGeometryParse Error: Can not match the Spatial Type.:" + eachRow.toString());
        return "";
    }

    //点数据支持3种格式数据的标准化
    //单字段wkt
    //双字段lon,lat
    //逗号隔开的单字段lon,lat
    private String getPointWKT(Map<ConstructedDataField, String> eachRow) {
        ConstructedDataField ptLon = null;
        ConstructedDataField ptLat = null;
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getFormat().equals(FieldFormatType.WKT_FIELD)) {
                return eachRow.get(field);
            }
            if (field.getFormat().equals(FieldFormatType.POINT_FIELD_CORS)) {
                try {
                    String[] lonlat = eachRow.get(field).split(",");
                    return lonlat2wktPoint(lonlat[0], lonlat[1]);
                } catch (Exception e) {
//                    logger.error("Unvalid cors:" + eachRow.get(field));
                    errors.add("Unvalid cors:" + eachRow.get(field));
                    return "";
                }
            }
            if (field.getFormat().equals(FieldFormatType.POINT_FIELD_LON))
                ptLon = field;
            if (field.getFormat().equals(FieldFormatType.POINT_FIELD_LAT))
                ptLat = field;
        }
        if (ptLat != null & ptLon != null) {
            return lonlat2wktPoint(eachRow.get(ptLon), eachRow.get(ptLat));
        }
//        logger.error("Can not get the Point By: WKT,Lon/Lat,Cors");
        errors.add("Can not get the Point By: WKT,Lon/Lat,Cors");
        return "";
    }

    //面数据支持2种格式数据的标准化
    //单字段wkt
    //单字段cors: lon1 lat1,lon2 lat2,lon3 lat3 ? 无法解析内环。
    private String getPolygonWKT(Map<ConstructedDataField, String> eachRow) {
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getFormat().equals(FieldFormatType.WKT_FIELD)) {
                return eachRow.get(field);
            }
            if (field.getFormat().equals(FieldFormatType.POLYGON_FIELD)) {
                return corsStr2wkt(FieldFormatType.POLYGON, eachRow.get(field));
            }
        }
//        logger.error("Can not get the Polygon By: WKT,Cors");
        errors.add("Can not get the Polygon By: WKT,Cors");
        return "";
    }

    //线数据支持2种格式数据的标准化
    //单字段wkt
    //单字段cors: lon1 lat1,lon2 lat2,lon3 lat3
    private String getPolyLineWKT(Map<ConstructedDataField, String> eachRow) {
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getFormat().equals(FieldFormatType.WKT_FIELD)) {
                return eachRow.get(field);
            }
            if (field.getFormat().equals(FieldFormatType.LINESTRING_FIELD)) {
                return corsStr2wkt(FieldFormatType.LINESTRING, eachRow.get(field));
            }
        }
//        logger.error("Can not get the Polygon By: WKT,Cors");
        errors.add("Can not get the Polygon By: WKT,Cors");
        return "";
    }

    //OD数据支持1种格式数据的标准化
    //四字段startLon,startLat,endLon,endLat
    private String getOdPairWKT(String fieldName, Map<ConstructedDataField, String> eachRow) {
        ConstructedDataField stLon = null;
        ConstructedDataField stLat = null;
        ConstructedDataField edLon = null;
        ConstructedDataField edLat = null;
        for (ConstructedDataField field : eachRow.keySet()) {
            switch (field.getFormat()) {
                case FieldFormatType.ODPAIR_FIELD_SLON:
                    stLon = field;
                    break;
                case FieldFormatType.ODPAIR_FIELD_SLAT:
                    stLat = field;
                    break;
                case FieldFormatType.ODPAIR_FIELD_ELON:
                    edLon = field;
                    break;
                case FieldFormatType.ODPAIR_FIELD_ELAT:
                    edLat = field;
                default:
                    break;
            }
        }
        switch (fieldName) {
            case FieldFormatType.SPATIAL_FIELD_NAME:
                if (stLon != null & stLat != null & edLon != null & edLat != null) {
                    return od2wktPolyLine(eachRow.get(stLon), eachRow.get(stLat), eachRow.get(edLon), eachRow.get(edLat));
                } else {
                    return "";
                }
            case FieldFormatType.SPATIAL_OD_ST_NAME:
                if (stLon != null & stLat != null) {
                    return lonlat2wktPoint(eachRow.get(stLon), eachRow.get(stLat));
                } else {
                    return "";
                }
            case FieldFormatType.SPATIAL_OD_ED_NAME:
                if (edLon != null & edLat != null) {
                    return lonlat2wktPoint(eachRow.get(edLon), eachRow.get(edLat));
                } else {
                    return "";
                }
            default:
                return "";
        }
    }

    private String lonlat2wktPoint(String lon, String lat) {
        try {
            return new GeometryFactory().createPoint(new Coordinate(Double.valueOf(lon), Double.valueOf(lat))).toText();
        } catch (Exception e) {
//            logger.error("Unvalid cors: " + lon + "," + lat);
            return "";
        }
    }

    private String corsStr2wkt(String type, String cors) {
        try {
            String[] corsArray = cors.split(",");
            Coordinate[] cords = new Coordinate[corsArray.length];
            for (int i = 0; i < corsArray.length; i++) {
                String[] lonlat = corsArray[i].split(" ");
                cords[i] = new Coordinate(Double.valueOf(lonlat[0]), Double.valueOf(lonlat[1]));
            }
            if (FieldFormatType.POLYGON.equals(type))
                return new GeometryFactory().createPolygon(cords).toText();
            if (FieldFormatType.LINESTRING.equals(type))
                return new GeometryFactory().createLineString(cords).toText();
            return "";
        } catch (Exception e) {
//            logger.error("Unvalid cors: " + cors);
            return "";
        }
    }

    private String od2wktPolyLine(String stLon, String stLat, String edLon, String edLat) {
        try {
            Coordinate[] cords = new Coordinate[2];
            cords[0] = new Coordinate(Double.valueOf(stLon), Double.valueOf(stLat));
            cords[1] = new Coordinate(Double.valueOf(edLon), Double.valueOf(edLat));
            return new GeometryFactory().createLineString(cords).toText();
        } catch (Exception e) {
//            logger.error("Unvalid od: " + stLon + stLat + edLat + edLat);
            return "";
        }
    }

}
